namespace StockSharp.Messages;

/// <summary>
/// Message channel, based on the queue and operate within a single process.
/// </summary>
public class InMemoryMessageChannel : Disposable, IMessageChannel
{
	private readonly IMessageQueue _queue;
	private readonly Action<Exception> _errorHandler;

	private CancellationTokenSource _cancellationTokenSource;
	private Task _processingTask;

	/// <summary>
	/// Initializes a new instance of the <see cref="InMemoryMessageChannel"/>.
	/// </summary>
	/// <param name="queue">Message queue.</param>
	/// <param name="name">Channel name.</param>
	/// <param name="errorHandler">Error handler.</param>
	public InMemoryMessageChannel(IMessageQueue queue, string name, Action<Exception> errorHandler)
	{
		if (name.IsEmpty())
			throw new ArgumentNullException(nameof(name));

		Name = name;

		_queue = queue ?? throw new ArgumentNullException(nameof(queue));
		_errorHandler = errorHandler ?? throw new ArgumentNullException(nameof(errorHandler));

		_queue.Close();
	}

	/// <summary>
	/// Handler name.
	/// </summary>
	public string Name { get; }

	/// <summary>
	/// Message queue count.
	/// </summary>
	public int MessageCount => _queue.Count;

	/// <summary>
	/// The channel cannot be opened.
	/// </summary>
	public bool Disabled { get; set; }

	private ChannelStates _state = ChannelStates.Stopped;

	/// <inheritdoc />
	public ChannelStates State
	{
		get => _state;
		private set
		{
			if (_state == value)
				return;

			_state = value;
			StateChanged?.Invoke();
		}
	}

	/// <inheritdoc />
	public event Action StateChanged;

	/// <inheritdoc />
	public void Open()
	{
		if (Disabled)
			return;

		State = ChannelStates.Started;
		_queue.Open();

		_cancellationTokenSource = new();

		_processingTask = ProcessMessagesAsync(_cancellationTokenSource.Token);
	}

	private async Task ProcessMessagesAsync(CancellationToken cancellationToken)
	{
		try
		{
			await foreach (var message in _queue.ReadAllAsync(cancellationToken).ConfigureAwait(false))
			{
				try
				{
					NewOutMessage?.Invoke(message);
				}
				catch (Exception ex)
				{
					_errorHandler(ex);
				}
			}
		}
		catch (OperationCanceledException)
		{
			// Normal cancellation
		}
		catch (Exception ex)
		{
			_errorHandler(ex);
		}
		finally
		{
			State = ChannelStates.Stopped;
		}
	}

	/// <inheritdoc />
	public void Close()
	{
		State = ChannelStates.Stopping;

		_cancellationTokenSource?.Cancel();
		_queue.Close();
		_queue.Clear();

		try
		{
			_processingTask?.Wait(TimeSpan.FromSeconds(5));
		}
		catch (AggregateException)
		{
			// Ignore cancellation exceptions
		}

		_cancellationTokenSource?.Dispose();
		_cancellationTokenSource = null;
	}

	void IMessageChannel.Suspend()
	{
		State = ChannelStates.Suspended;
	}

	void IMessageChannel.Resume()
	{
		State = ChannelStates.Started;
	}

	void IMessageChannel.Clear()
	{
		_queue.Clear();
	}

	/// <inheritdoc />
	public bool SendInMessage(Message message)
	{
		if (!this.IsOpened())
		{
			//throw new InvalidOperationException();
			return false;
		}

		_queue.Enqueue(message);

		return true;
	}

	/// <inheritdoc />
	public event Action<Message> NewOutMessage;

	/// <summary>
	/// Create a copy of <see cref="InMemoryMessageChannel"/>.
	/// </summary>
	/// <returns>Copy.</returns>
	public virtual IMessageChannel Clone()
		=> new InMemoryMessageChannel(_queue, Name, _errorHandler);

	object ICloneable.Clone() => Clone();

	/// <inheritdoc />
	protected override void DisposeManaged()
	{
		Close();
		base.DisposeManaged();
	}
}